package com.cxyd.client;

import com.cxyd.common.ClientConstants;
import com.cxyd.handler.TransferHandler;
import com.cxyd.utils.AssertUtil;
import com.cxyd.utils.MyDecoder;
import com.cxyd.utils.MyEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class TransportClient {
    private static final ScheduledExecutorService SCHEDULER = Executors.newScheduledThreadPool(1,
            Executors.defaultThreadFactory());
    //ip与端口号
    private final String host;
    private final int port;
    private final String message;
    //重连间隔时间
    public static final long RECONNECT_DELAY_MS = 2000L;
    //定义Channel、Group与Handler
    private Channel channel;

    private NioEventLoopGroup eventLoopGroup;

    private TransferHandler clientHandler;

    private final AtomicInteger currentState = new AtomicInteger(ClientConstants.CLIENT_STATUS_OFF);
    private final AtomicBoolean shouldRetry = new AtomicBoolean(true);
    private final AtomicInteger failConnectedTime = new AtomicInteger(0);

    public TransportClient(String host, int port, String message) {
        //ip与端口号不能为空
        AssertUtil.notEmpty(host, "remote host cannot be blank");
        AssertUtil.isTrue(port > 0, "port should be positive");
        this.host = host;
        this.port = port;
        this.message = message;
    }

    /**
     * 初始化Bootstrap引导类对象
     */
    private Bootstrap initClientBootstrap() {
        Bootstrap b = new Bootstrap();
        eventLoopGroup = new NioEventLoopGroup();
        b.group(eventLoopGroup)  //设置线程组
                .channel(NioSocketChannel.class)  //设置客户端通道的实现类（反射方式）
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, ClientConstants.DEFAULT_CONNECT_TIMEOUT_MILLIS)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        clientHandler = new TransferHandler(currentState, disconnectCallback, message);
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("myEncode", new MyEncoder());
                        pipeline.addLast("myDecode", new MyDecoder());
                        pipeline.addLast(clientHandler);//加入自定义的处理器
                    }
                });
        return b;
    }

    //长连接
    private void connect(Bootstrap b) {
        if (currentState.compareAndSet(ClientConstants.CLIENT_STATUS_OFF, ClientConstants.CLIENT_STATUS_PENDING)) {
            b.connect(host, port)
                    .addListener((GenericFutureListener<ChannelFuture>) future -> {
                        if (future.cause() != null) {
                            log.warn(
                                    String.format("[ty_client] Could not connect to <%s:%d> after %d times",
                                            host, port, failConnectedTime.get()), future.cause());
                            failConnectedTime.incrementAndGet();
                            channel = null;
                        } else {
                            failConnectedTime.set(0);
                            channel = future.channel();
                            log.info("[ty_client] Successfully connect to server <{}:{}>", host, port);
                        }
                    });
        }
    }

    /**
     * 通讯重连(守护线程)
     */
    private Runnable disconnectCallback = new Runnable() {
        @Override
        public void run() {
            if (!shouldRetry.get()) {
                return;
            }
            SCHEDULER.schedule(() -> {
                if (shouldRetry.get()) {
                    log.info("[ty_client] Reconnecting to server <{}:{}>", host, port);
                    try {
                        startInternal();
                    } catch (Exception e) {
                        log.warn("[ty_client] Failed to reconnect to server", e);
                    }
                }
            }, RECONNECT_DELAY_MS * (failConnectedTime.get() + 1), TimeUnit.MILLISECONDS);
            cleanUp();
        }
    };

//    private Runnable sendRequest

    /**
     * 暴露给外层调用的方法
     */
    public void start() throws Exception {
        shouldRetry.set(true);
        startInternal();
    }

    private void startInternal() {
        connect(initClientBootstrap());
    }

    private void cleanUp() {
        if (channel != null) {
            channel.close();
            channel = null;
        }
        if (eventLoopGroup != null) {
            eventLoopGroup.shutdownGracefully();
        }
    }

    public boolean isReady() {
        return channel != null && clientHandler != null && clientHandler.hasStarted();
    }
}
